package ar.edu.unq.sandsoft.session;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import javax.xml.parsers.ParserConfigurationException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.xml.sax.SAXException;

import ar.edu.unq.sandsoft.cache.SandCacheImpl;
import ar.edu.unq.sandsoft.constructors.SandFileConfigReader;
import ar.edu.unq.sandsoft.constructors.SandFileReader;
import ar.edu.unq.sandsoft.exceptions.CouldnFindAttribute;
import ar.edu.unq.sandsoft.utils.ParsingHelper;
import ar.edu.unq.sandsoft.utils.SandFile;

public class SandSessionFactory {
    private static String MACHINENAME;

    private static SandCacheImpl cache;

    private static List<SandFileReader> readers = new ArrayList<SandFileReader>();

    public static SandSession createSession() throws SAXException, IOException, ParserConfigurationException,
            CouldnFindAttribute {
        loadConfiguration();
        createTables();
        return new SandSessionImpl(readers, getCache());
    }

    private static void createTables() throws IOException, CouldnFindAttribute, SAXException,
            ParserConfigurationException {
        HBaseAdmin admin = connectToAdmin();
        for (SandFileReader reader : SandSessionFactory.getReaders()) {
            reader.glance();
            createTable(admin, reader);
        }
    }

    public static void dropAllTables() {
        try {
            loadConfiguration();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        HBaseAdmin admin = connectToAdmin();
        for (HTableDescriptor tableDescriptor : getTables()) {
            String tableName = tableDescriptor.getNameAsString();
            try {
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            } catch (IOException e) {
                throw new UnsupportedOperationException();
            }
        }
    }

    private static SandCacheImpl getCache() {
        if (cache == null) {
            cache = new SandCacheImpl();
        }
        return cache;
    }

    static SandFileReader getReaderByClassName(String className) {
        SandFileReader wantedReader = null;
        for (SandFileReader reader : readers) {
            if (reader.findClassName().equals(className))
                wantedReader = reader;
        }
        wantedReader.lookupKey();
        return wantedReader;
    }

    private static void loadClassFile(String string) {
        getReaders().add(new SandFileReader(new SandFile(string + ".xml")));
    }

    public static void loadConfiguration() throws SAXException, IOException, ParserConfigurationException,
            CouldnFindAttribute {
        SandFileConfigReader configurationReader = new SandFileConfigReader();
        for (String config : configurationReader.loadConfiguration()) {
            loadClassFile(config);
        }
        MACHINENAME = configurationReader.loadServer();
    }

    public static boolean areTablesEmpty() {
        return getTables().isEmpty();
    }

    private static List<HTableDescriptor> getTables() {
        try {
            return Arrays.asList(connectToAdmin().listTables());
        } catch (IOException e) {
            throw new UnsupportedOperationException();
        }
    }

    public static HBaseAdmin connectToAdmin() {
        try {
            Configuration config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", MACHINENAME);
            return new HBaseAdmin(config);
        } catch (MasterNotRunningException e) {
            throw new RuntimeException("Could not setup HBaseAdmin as no master is running, did you start HBase?...", e);
        } catch (ZooKeeperConnectionException e) {
            throw new RuntimeException("Error while connecting to HBase, Zookeeper connection failed", e);
        }
    }

    protected static void createTable(HBaseAdmin admin, SandFileReader reader) throws IOException, CouldnFindAttribute {
        String className = new ParsingHelper().adaptClassName(reader.findClassName());
        if (!admin.tableExists(className)) {
            admin.createTable(new HTableDescriptor(className));
            admin.disableTable(className);

            for (String attributeName : reader.findAttributes().keySet()) {
                admin.addColumn(className, new HColumnDescriptor(attributeName));
            }

            admin.enableTable(className);
        }
    }

    public static void setReaders(List<SandFileReader> readers) {
        SandSessionFactory.readers = readers;
    }

    public static List<SandFileReader> getReaders() {
        return readers;
    }
}
